package com.yesun.task.pipeline.handler.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yesun.core.util.FileUtils;
import com.yesun.server.mdp.service.client.impl.DefaultClient;
import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.common.StockCode;
import com.yesun.task.pipeline.domain.dto.JobDto;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Handler;
import com.yesun.task.pipeline.handler.Loader;
import com.yesun.task.pipeline.service.impl.ClickhouseServiceImpl;
import com.yesun.task.pipeline.sink.ClickhouseSink;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.BlockingQueue;

//实时：分时分笔数据Handler
@Slf4j
public class RtDetailHandler extends Handler {

    static DefaultClient client = new DefaultClient();
    static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    static SimpleDateFormat shortSdf = new SimpleDateFormat("yyyyMMdd");
    static String TODAY = shortSdf.format(Calendar.getInstance().getTime());
    static Loader loader;

    public RtDetailHandler(JobDto jobDto) {
        super(jobDto);
    }

    @Override
    public Loader getLoader() {
        if(loader == null) {
            //实例化
            loader = new Loader() {
                @Override
                public void load(BlockingQueue queue) {
                    log.info("加载分时分笔子任务...");
                    //实现资金流入子任务，插入到队列
                    String[] arr = StockCode.STOCK_CODE_YOUXIU.split(",");
                    for(String code : arr) {
                        //update
                        long currentTime = System.currentTimeMillis();
                        String secid = "";
                        if(code.startsWith("6")) {
                            secid = "1." + code;
                        } else if(code.startsWith("0") || code.startsWith("3") || code.startsWith("8")) {
                            secid = "0." + code;
                        } else {
                            //暂时不支持其他股票
                            log.info("暂时不支持的股票类型：" + code);
                            continue;
                        }
                        //TODO 需要访问其他服务器
                        Random rand = new Random();
                        String url = "https://"+rand.nextInt(5000)+".push2.eastmoney.com/api/qt/stock/details/sse?fields1=f1%2Cf2%2Cf3%2Cf4&fields2=f51%2Cf52%2Cf53%2Cf54%2Cf55&mpi=2000&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&pos=-0&secid="+secid+"&wbp2u=%7C0%7C0%7C0%7Cweb";
                        TaskDto taskDto = new TaskDto();
                        taskDto.setJobName(jobDto.getJobName());
                        taskDto.setName(code + "分时分笔实时采集任务");
                        taskDto.setTag(Constant.JOB_TAG_RT_DETAIL);

                        Map<String,String> map = new HashMap<>();
                        map.put("url", url);
                        map.put("code", code);
                        map.put("saveTo", "/df_detail/"+ shortSdf.format(new Date()) +"/"+code+".csv");//暂时只支持保存到磁盘
                        taskDto.setMap(map);

                        queue.offer(taskDto);
                        log.info("添加子任务：" + taskDto.getName());
                    }
                }
            };
        }
        return loader;
    }

    @Override
    public Executor getExecutor() {
        return new Executor() {
            @Override
            public void execute(TaskDto taskDto) {
                //执行task的采集
                log.info("execute：" + taskDto.getName());
                long start = System.currentTimeMillis();

                ClickhouseServiceImpl clickhouseServiceImpl = (ClickhouseServiceImpl) com.yesun.task.pipeline.common.SpringUtil.APPLICATION_CONTEXT.getBean("clickhouseServiceImpl");

                //从数据库读取当前ts_code的最新记录时间戳
                String timestamp = "";
                List<Map<String, Object>> list = clickhouseServiceImpl.query("SELECT ts_code, trade_datetime FROM ods_stock_rt_detail_df WHERE dt='"+TODAY+"' AND ts_code='"+taskDto.getMap().get("code")+"' ORDER BY trade_datetime DESC limit 1");
                if(list != null &&  list.size()>0) {
                    Map<String,Object> map = list.get(0);
                    log.info(map.get("trade_datetime").toString());
                    timestamp = map.get("trade_datetime").toString();
                }

                try {
//                    StringBuffer stringBuffer = new StringBuffer();
                    //由于数据源是一个event-stream接口，所以单独实现
                    CloseableHttpClient httpclient = HttpClients.createDefault();
                    CloseableHttpResponse response = null;
                    HttpGet httpGet = new HttpGet(taskDto.getMap().get("url"));
                    try {
                        response = httpclient.execute(httpGet);
                        HttpEntity entity = response.getEntity();
                        Header[] headers = response.getAllHeaders();
                        if (entity != null) {

                            log.info(taskDto.getMap().get("url") + "success");

                            InputStream inputStream = entity.getContent();
                            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                            String readline;
                            int index = 0;
                            while ((readline = reader.readLine()) != null) {
                                // 处理 event stream 数据
//                                stringBuffer.append(readline);
                                if(readline.indexOf("{") == -1) {
                                    continue;
                                }
                                readline = readline.substring(readline.indexOf("{"), readline.lastIndexOf("}")+1);
                                StringBuffer sb = new StringBuffer("采集日期,股票代码,时间,价格,交易量,笔数,交易类型\n"); //股价最好从其他表获取
                                JSONObject objRoot = JSONObject.parseObject(readline);
                                if(objRoot != null) {
                                    JSONObject objData = objRoot.getJSONObject("data");
                                    if(objData != null) {
                                        JSONArray jsonArray = objData.getJSONArray("details");
                                        if (jsonArray != null) {
                                            index++;
                                            //判断长度
                                            if(index == 1) {
                                                //表示第一条，只需要处理时间戳之后的
                                                if(StringUtils.isNotBlank(timestamp)){
                                                    boolean started = false;
                                                    for(int i=0;i<jsonArray.size();i++) {
                                                        String[] arr = jsonArray.get(i).toString().split(",");
                                                        if(started == false) {
                                                            if(arr[0].equals(timestamp)) {
                                                                started = true;
                                                            }
                                                        } else {
                                                            ////写入到缓存队列，等待批量写入到ck
//                                                            log.info("写入Queue....");
                                                            String msg = TODAY + "," + taskDto.getMap().get("code") + "," + arr[0] + "," + arr[1] + "," +  arr[2] + "," + arr[3] + "," + arr[4];
                                                            ClickhouseSink.QUEUE.offer(msg);
//                                                            log.info(jsonArray.get(i).toString());
                                                        }
                                                    }
                                                } else {
                                                    for(int i=0;i<jsonArray.size();i++) {
//                                                        log.info(jsonArray.get(i).toString());
                                                        String[] arr = jsonArray.get(i).toString().split(",");
//                                                        log.info("写入Queue....");
//                                                        log.info(jsonArray.get(i).toString());
                                                        String msg = TODAY + "," + taskDto.getMap().get("code") + "," + arr[0] + "," + arr[1] + "," +  arr[2] + "," + arr[3] + "," + arr[4];
                                                        ClickhouseSink.QUEUE.offer(msg);
                                                    }
                                                }
                                            } else {
                                                jsonArray.stream()
                                                        .forEach(line -> {
                                                            String[] arr = line.toString().split(",");
//                                                            log.info("写入Queue....");
//                                                            log.info(line.toString());
                                                            String msg = TODAY + "," + taskDto.getMap().get("code") + "," + arr[0] + "," + arr[1] + "," +  arr[2] + "," + arr[3] + "," + arr[4];
                                                            ClickhouseSink.QUEUE.offer(msg);
                                                        });
                                            }
                                        } else {
//                                            log.info("details is empty:" + taskDto.getMap().get("url"));
//                                            log.info(readline);
                                        }
                                    } else {
//                                        log.info("data is empty！");
//                                        log.info(readline);
                                    }
                                } else {
//                                    log.info("content is empty！");
//                                    log.info(readline);
                                }
                            }
                        }
                    } catch (org.apache.http.client.ClientProtocolException e) {
                        log.info(taskDto.getMap().get("url") + " fail");
                        e.printStackTrace();
                    } catch (IOException e) {
                        log.info(taskDto.getMap().get("url") + " fail");
                        e.printStackTrace();
                    } finally {
                        if(response!=null) {
                            try {
                                response.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }
}
